package org.example.mq.mqsever.datacenter;

/**
 * Created with IntelliJ IDEA.
 * Description:做渐进主义的改良派,不做完没注意的速成派
 * User: XALEXC
 * Date: 2025-05-18
 * Time: 15:58
 */

import org.example.mq.common.MqException;
import org.example.mq.mqsever.core.Binding;
import org.example.mq.mqsever.core.Exchange;
import org.example.mq.mqsever.core.MSGQueue;
import org.example.mq.mqsever.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Exchanger;

/**
 * 使用这个类来统一管理内存中的所有数据
 * 该类后续提供的一些方法,可能会在多线程环境下被使用.因此要注意线程安全问题
 */
public class MemoryDataCenter {
    // key是exchangeName,value是Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key是queueName,value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue>queueMap = new ConcurrentHashMap<>();
    // 第一个key是exchangeName,第二个key是queueName
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap=new ConcurrentHashMap<>();
    // key是messageId,value是Message对象
    private ConcurrentHashMap<String, Message>messageMap=new ConcurrentHashMap<>();
    // key是queueName,value是一个Message的链表
    private ConcurrentHashMap<String, LinkedList<Message>>queueMessageMap=new ConcurrentHashMap<>();
    // 第一个key是queueName,第二个key是MessageId
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>>queueMessageWaitAckMap=new ConcurrentHashMap<>();

    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
        System.out.println("[MemoryDataCenter] 新交换机添加成功! exchangeName="+exchange.getName());
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName="+exchangeName);
    }

    public void insertQueue(MSGQueue queue){
        queueMap.put(queue.getName(),queue);
        System.out.println("[MemoryDataCenter] 新队列添加成功! queueName="+queue.getName());
    }

    public MSGQueue getQueue(String queueName){
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName){
        queueMap.remove(queueName);
        System.out.println("[MemoryDataCenter] 队列删除成功! queueName="+queueName);
    }

    public void insertBinding(Binding binding) throws MqException {
        // 先使用exchangeName查一下,对应的哈希表是否存在
        ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k->new ConcurrentHashMap<>());
      synchronized (bindingMap){
          if (bindingMap.get(binding.getQueueName())!=null){
              throw new MqException("[MemoryDataCenter] 绑定已存在!exchangeName="+binding.getExchangeName()+
                      ", queueName="+binding.getQueueName());
          }
          bindingMap.put(binding.getQueueName(),binding);
          System.out.println("[MemoryDataCenter] 新绑定添加成功! exchangeName="+binding.getExchangeName()
          + ", queueName="+binding.getQueueName());
      }
    }

    // 获取绑定,写两个版本
    // 1.根据 exchangeName 和 queueName 确定唯一一个 Binding
    // 2.根据 exchangeName 获取到所有的 Binding
    public Binding getBinding(String queueName,String exchangeName) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.get(exchangeName);
        if (bindingMap==null){
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.get(binding.getExchangeName());
        if (bindingMap==null){
            // 该交换机没有绑定任何队列
            throw new MqException("[MemoryDataCenter] 绑定不存在! exchangeName="+binding.getExchangeName()
            +", queueName="+binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("[MemoryDataCenter] 绑定删除成功! exchangeName="+binding.getExchangeName()
                + ", queueName="+binding.getQueueName());
    }

    // 添加消息
    public void addMessage(Message message) throws MqException {
        messageMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 新消息添加成功! messageId="+message.getMessageId());
    }

    // 根据 ID 查询消息
    public Message getMessage(String messageId) throws MqException {
        return messageMap.get(messageId);
    }

    // 根据 id 删除消息
    public void removeMessage(String messageId) throws MqException {
        messageMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息删除成功! messageId="+messageId);

    }

    // 发送消息到制定队列
    public void sendMessage(MSGQueue queue,Message message) throws MqException {
        // 把消息放到对应队列的数据结构当中
        // 先根据队列的名字,找到该队列对应的消息链表
        LinkedList<Message> messages=queueMessageMap.computeIfAbsent(queue.getName(),
                k->new LinkedList<>());
        // 再把数据加到message里面
        synchronized (messages){
            messages.add(message);
        }
        // 在这里吧该消息也往消息中心中插入一下,假设如果message已经在消息中心中存在,
        // 重复插入也没关系
        // 主要就是相同 messageId,对应的 message 的内容一定是一样的.(服务器代码不会得Message内容做修改basicProperties和body)
        addMessage(message);
        System.out.println("[MemoryDataCenter] 消息被投递到队列中! messageId="+message.getMessageId());
    }

    // 从队列中取消息
    public Message pollMessage(String queueName){
        // 根据队列名,查找一下,对应的队列的消息链表
        LinkedList<Message> messages=queueMessageMap.get(queueName);
        if (messages==null){
            return null;
        }
        synchronized (messages){
            if(messages.size()==0){
                return null;
            }
            Message currentMessage=messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId="+currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取指定队列中消息的个数
    public int getMessageCount(String queueName){
        LinkedList<Message> message=queueMessageMap.get(queueName);
        if (message==null){
            return 0;
        }
        synchronized (message){
            return message.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName,Message message){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.computeIfAbsent(queueName,
                k->new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(),message);
        System.out.println("[MemoryDataCenter] 消息进入待确认队列! messageId="+message.getMessageId());
    }

    // 删除未确认的消息(消息已经确认了)
    public void removeMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if (messageHashMap==null){
            return;
        }
        messageHashMap.remove(messageId);
        System.out.println("[MemoryDataCenter] 消息从待确认队列删除! messageId="+messageId);
    }

    // 获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,String messageId){
        ConcurrentHashMap<String,Message> messageHashMap=queueMessageWaitAckMap.get(queueName);
        if (messageHashMap==null){
            return null;
        }
        return messageHashMap.get(messageId);
    }

    // 这个方法就是从硬盘上读取数据,把硬盘中之前的持久化存储的各个维度的数据都恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        // 清空之前的所有数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 恢复所有的交换机的数据
        List<Exchange> exchanges=diskDataCenter.selectAllExchanges();
        for (Exchange exchange:exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        // 恢复所有的队列数据
        List<MSGQueue> queues=diskDataCenter.selectAllQueues();
        for (MSGQueue queue:queues){
            queueMap.put(queue.getName(),queue);
        }
        // 恢复所有绑定数据
        List<Binding> bindings=diskDataCenter.selectAllBindings();
        for (Binding binding:bindings){
            ConcurrentHashMap<String,Binding> bindingMap=bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }

        // 恢复所有的消息数据
        // 遍历所有的队列,根据每个消息的名字,获取到所有的消息
        for (MSGQueue queue:queues){
            LinkedList<Message> messages=diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(),messages);
            for (Message message:messages){
                messageMap.put(message.getMessageId(),message);
            }
        }
        // 注意!针对"未确认的消息"这部分内存中的数据,不需要从硬盘恢复.之前考虑硬盘存储的时候,也没设定这一块
        // 一旦在等待ack的过程中,服务器重启了,此时这些"未被确认的消息",就恢复成"未被取走的消息"
        // 这个消息在硬盘上存储的时候,就是当做"未被取走"
    }
}
